import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.*;
import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.List;

public class TestAPI2 {
    Connection conn;
    Admin admin;

    @Before
    public void createConn() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
    }
    @Test
    public void createtable() throws IOException {
        HTableDescriptor student = new HTableDescriptor (TableName.valueOf ("student"));
        HColumnDescriptor info = new HColumnDescriptor ("info");
        student.addFamily (info);
        admin.createTable (student);

    }
    @Test
    public void dropTable() throws IOException {
        TableName student = TableName.valueOf ("student");
        if (admin.tableExists (student)){
            admin.disableTable (student);
            admin.deleteTable (student);
        }
    }
    @Test
    public void List() throws IOException {
        TableName[] tableNames = admin.listTableNames ( );
        for (TableName name : tableNames) {
            System.out.println (name.getNameAsString () );
        }
    }
    @Test
    public void get() throws IOException {
        Table student = conn.getTable (TableName.valueOf ("student"));
        Get get1 = new Get ("1500100009".getBytes ( ));
        Result rs = student.get (get1);
        byte[] rk = rs.getRow ( );
        System.out.println (rk );
        System.out.println (Bytes.toString (rk) );
        byte[] name = rs.getValue ("info".getBytes ( ), "name".getBytes ( ));
        System.out.println (name );
        System.out.println (Bytes.toString (name) );

    }
    @Test
    public void putAll() throws IOException {
        /**
         * 读取学生信息数据
         */

        BufferedReader br = new BufferedReader(new FileReader("H:\\soft\\IDEA1.8\\bd13\\Hbase\\src\\students.txt"));

        // 与HBase中的student表建立连接
        Table student = conn.getTable(TableName.valueOf("student"));

        String line = null;

        // 创建Put的集合
        ArrayList<Put> puts = new ArrayList();
        int batchSize = 11;
        while ((line = br.readLine()) != null) {
            // 写入HBase
            String[] splits = line.split(",");
            String id = splits[0];
            String name = splits[1];
            String age = splits[2];
            String gender = splits[3];
            String clazz = splits[4];

            Put put = new Put(id.getBytes());
            byte[] info = "info".getBytes();
            put.addColumn(info, "name".getBytes(), name.getBytes());
            put.addColumn(info, "age".getBytes(), age.getBytes());
            put.addColumn(info, "gender".getBytes(), gender.getBytes());
            put.addColumn(info, "clazz".getBytes(), clazz.getBytes());


            // 每条数据都会执行一次，效率很慢
//            student.put(put);

            // 将每个Put对象加入puts集合
            puts.add(put);
            // 当puts集合的大小同batchSize大小一致时，则调用HTable的put方法进行批量写入
            if (puts.size() == batchSize) {
                student.put(puts);
                // 清空集合
                puts.clear();
            }

        }
        System.out.println(puts.isEmpty());
        System.out.println(puts.size());
        // 当batchSize的大小同数据的条数不成整比的时候 可能会造成最后几条数据未被写入
        // 手动去判断puts集合是否为空，不为空则将其写入HBase
        if (!puts.isEmpty()) {
            student.put(puts);
        }


        br.close();


    }
    @Test
    public void scantable() throws IOException {
        Table student = conn.getTable (TableName.valueOf ("student"));
        Scan scan = new Scan ( );
//        scan.withStartRow ("1500100100".getBytes ());
//        scan.withStopRow ("1500100111".getBytes ());
//        scan.setLimit (10);
        for (Result result : student.getScanner (scan)) {
            String id = Bytes.toString (result.getRow ( ));
            String name = Bytes.toString (result.getValue ("info".getBytes ( ), "name".getBytes ( )));
            String age = Bytes.toString (result.getValue ("info".getBytes ( ), "age".getBytes ( )));
            String gender = Bytes.toString (result.getValue ("info".getBytes ( ), "gender".getBytes ( )));
            String clazz = Bytes.toString (result.getValue ("info".getBytes ( ), "clazz".getBytes ( )));
            System.out.println (id+","+name+"+"+age+","+gender+","+clazz );
        }

    }
    @Test
    public void scancell() throws IOException {
        Table student = conn.getTable (TableName.valueOf ("student"));
        Scan scan = new Scan ( );
        scan.withStopRow ("1500100990".getBytes ());
        ResultScanner scanner = student.getScanner (scan);
        for (Result result : scanner) {
            String id = Bytes.toString (result.getRow ( ));
            System.out.println (id+" " );
            List<Cell> cells = result.listCells ( );
            for (Cell cell : cells) {
                String value = Bytes.toString (CellUtil.cloneValue (cell));
                System.out.print (value+" " );

            }
            System.out.println ( );
        }
    }
    @After
    public void close() throws IOException {
        admin.close ();
        conn.close ();

    }

}
